package com.abyss.transformation;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.operators.PartitionOperator;
import org.apache.flink.api.java.tuple.Tuple2;
/**
 * @author Abyss
 * @date 2020/10/4
 * @description 按照指定列的hash值进行分区
 */
public class PartitionByHashDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(2);
        // 2. source
        DataSource<Tuple2<Integer, Integer>> source = env.fromElements(
                Tuple2.of(1, 1),
                Tuple2.of(2, 1),
                Tuple2.of(3, 1),
                Tuple2.of(1, 1),
                Tuple2.of(2, 1),
                Tuple2.of(3, 1),
                Tuple2.of(1, 1),
                Tuple2.of(2, 1),
                Tuple2.of(3, 1),
                Tuple2.of(1, 1),
                Tuple2.of(4, 1),
                Tuple2.of(5, 1)
        );

        // 3. 进行hash分区
        PartitionOperator<Tuple2<Integer, Integer>> partitionedDataSet = source.partitionByHash(0);

        // 4. 进行分区号的打印
        MapOperator<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> map = partitionedDataSet.map(new RichMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception {
                return Tuple2.of(value.f0, getRuntimeContext().getIndexOfThisSubtask());
            }
        });

        // 5. print
        map.print();
    }
}
